package messaging

import (
	"context"
	"database/sql"
	"fmt"
	"log"
	"time"

	"github.com/IBM/sarama"
	"github.com/ev-charging/cloud-platform/config"
	"github.com/go-redis/redis/v8"
)

// 添加 Redis 导入

// KafkaProducer Kafka生产者封装
type KafkaProducer struct {
	producer sarama.SyncProducer
	config   *config.KafkaConfig
}

// KafkaConsumer Kafka消费者封装
type KafkaConsumer struct {
	consumer sarama.Consumer
	config   *config.KafkaConfig
	groupID  string
	topics   []string
}

// InitKafkaProducer 初始化Kafka生产者
func InitKafkaProducer(brokers []string) (*KafkaProducer, error) {
	// 创建配置
	config := sarama.NewConfig()
	config.Producer.RequiredAcks = sarama.WaitForAll
	config.Producer.Retry.Max = 3
	config.Producer.Return.Successes = true
	config.Producer.Return.Errors = true
	config.Producer.Compression = sarama.CompressionSnappy
	config.Producer.Flush.Frequency = 500 * time.Millisecond
	config.Version = sarama.V2_8_1_0

	// 创建同步生产者
	producer, err := sarama.NewSyncProducer(brokers, config)
	if err != nil {
		return nil, fmt.Errorf("创建Kafka生产者失败: %w", err)
	}

	return &KafkaProducer{
		producer: producer,
	}, nil
}

// SendMessage 发送消息到指定主题
func (p *KafkaProducer) SendMessage(topic string, key, value string) (partition int32, offset int64, err error) {
	msg := &sarama.ProducerMessage{
		Topic: topic,
		Key:   sarama.StringEncoder(key),
		Value: sarama.StringEncoder(value),
	}

	partition, offset, err = p.producer.SendMessage(msg)
	if err != nil {
		return 0, 0, fmt.Errorf("发送Kafka消息失败: %w", err)
	}

	return partition, offset, nil
}

// CloseKafkaProducer 关闭Kafka生产者
func CloseKafkaProducer(p *KafkaProducer) error {
	if p != nil && p.producer != nil {
		return p.producer.Close()
	}
	return nil
}

// InitKafkaConsumer 初始化Kafka消费者
func InitKafkaConsumer(brokers []string, groupID string) (*KafkaConsumer, error) {
	// 创建配置
	config := sarama.NewConfig()
	config.Consumer.Return.Errors = true
	config.Version = sarama.V2_8_1_0

	// 创建消费者
	consumer, err := sarama.NewConsumer(brokers, config)
	if err != nil {
		return nil, fmt.Errorf("创建Kafka消费者失败: %w", err)
	}

	return &KafkaConsumer{
		consumer: consumer,
		groupID:  groupID,
	}, nil
}

// ConsumeMessages 消费指定主题的消息
func (c *KafkaConsumer) ConsumeMessages(topic string, partition int32, offset int64, handler func(msg *sarama.ConsumerMessage) error) error {
	pc, err := c.consumer.ConsumePartition(topic, partition, offset)
	if err != nil {
		return fmt.Errorf("消费分区失败: %w", err)
	}
	defer pc.Close()

	for {
		select {
		case msg := <-pc.Messages():
			if err := handler(msg); err != nil {
				log.Printf("处理消息失败: %v", err)
			}
		case err := <-pc.Errors():
			log.Printf("Kafka消费错误: %v", err)
			return err
		}
	}
}

// CloseKafkaConsumer 关闭Kafka消费者
func CloseKafkaConsumer(c *KafkaConsumer) error {
	if c != nil && c.consumer != nil {
		return c.consumer.Close()
	}
	return nil
}

// MessageHandler 消息处理器接口
type MessageHandler interface {
	HandleDeviceData(ctx context.Context, deviceID string, data map[string]interface{}) error
	HandleTransaction(ctx context.Context, transactionID string, data map[string]interface{}) error
}

// KafkaMessageWorker Kafka消息处理工作器
type KafkaMessageWorker struct {
	consumer *KafkaConsumer
	handler  MessageHandler
	topic    string
}

// NewKafkaMessageWorker 创建新的消息处理工作器
func NewKafkaMessageWorker(consumer *KafkaConsumer, handler MessageHandler, topic string) *KafkaMessageWorker {
	return &KafkaMessageWorker{
		consumer: consumer,
		handler:  handler,
		topic:    topic,
	}
}

// Start 启动消息处理工作器
func (w *KafkaMessageWorker) Start() error {
	// 获取分区列表
	partitions, err := w.consumer.consumer.Partitions(w.topic)
	if err != nil {
		return fmt.Errorf("获取分区列表失败: %w", err)
	}

	// 为每个分区启动一个消费者协程
	for _, partition := range partitions {
		go func(p int32) {
			for {
				err := w.consumer.ConsumeMessages(w.topic, p, sarama.OffsetNewest, w.processMessage)
				if err != nil {
					log.Printf("分区 %d 消费失败: %v, 5秒后重试...", p, err)
					time.Sleep(5 * time.Second)
				}
			}
		}(partition)
	}

	return nil
}

// processMessage 处理单个消息
func (w *KafkaMessageWorker) processMessage(msg *sarama.ConsumerMessage) error {
	// 这里应该根据消息类型路由到不同的处理器方法
	// 简化实现，实际项目中应该有更复杂的路由逻辑
	ctx := context.Background()

	// 假设消息是JSON格式，这里需要实际解析
	// 简化处理，直接调用处理器接口
	switch w.topic {
	case "device-data":
		return w.handler.HandleDeviceData(ctx, string(msg.Key), map[string]interface{}{
			"value":     string(msg.Value),
			"timestamp": msg.Timestamp,
		})
	case "transactions":
		return w.handler.HandleTransaction(ctx, string(msg.Key), map[string]interface{}{
			"value":     string(msg.Value),
			"timestamp": msg.Timestamp,
		})
	default:
		return fmt.Errorf("未知主题: %s", w.topic)
	}
}

// KafkaMessageHandler Kafka消息处理器
type KafkaMessageHandler struct {
	consumer *KafkaConsumer
	db       *sql.DB
	// 需确保导入 redis 包，此处使用 github.com/go-redis/redis/v8
	redisClient *redis.Client
	ctx         context.Context
	cancel      context.CancelFunc
}

// NewKafkaMessageHandler 创建新的Kafka消息处理器
func NewKafkaMessageHandler(consumer *KafkaConsumer, db *sql.DB, redisClient *redis.Client) *KafkaMessageHandler {
	ctx, cancel := context.WithCancel(context.Background())
	return &KafkaMessageHandler{
		consumer:    consumer,
		db:          db,
		redisClient: redisClient,
		ctx:         ctx,
		cancel:      cancel,
	}
}

// Start 启动消息处理
func (h *KafkaMessageHandler) Start() error {
	// 这里实现实际的消息处理逻辑
	log.Println("Kafka消息处理器启动")

	// 这里实现实际的消息处理逻辑
	// 示例：处理设备数据主题
	go h.consumeTopic("device-data", h.handleDeviceData)
	go h.consumeTopic("transactions", h.handleTransaction)

	return nil
}

// consumeTopic 消费指定主题的消息
func (h *KafkaMessageHandler) consumeTopic(topic string, handler func(msg *sarama.ConsumerMessage)) {
	// 获取主题分区列表
	partitions, err := h.consumer.consumer.Partitions(topic)
	if err != nil {
		log.Printf("获取分区列表失败: %v", err)
		return // 直接返回而不是继续重试
	}

	// 为每个分区创建消费者
	for _, partition := range partitions {
		go func(p int32) {
			// 创建独立的分区消费者
			pc, err := h.consumer.consumer.ConsumePartition(topic, p, sarama.OffsetNewest)
			if err != nil {
				log.Printf("消费分区 %d 失败: %v", p, err)
				return // 失败后不再重试
			}
			defer pc.Close()

			for {
				select {
				case msg := <-pc.Messages():
					handler(msg)
				case err := <-pc.Errors():
					log.Printf("分区 %d 消费错误: %v", p, err)
				case <-h.ctx.Done():
					log.Printf("分区 %d 消费停止", p)
					return
				}
			}
		}(partition)
	}
}

// 添加消息处理函数
func (h *KafkaMessageHandler) handleDeviceData(msg *sarama.ConsumerMessage) {
	// 实际处理设备数据的逻辑
	log.Printf("处理设备数据: Topic=%s, Key=%s", msg.Topic, string(msg.Key))

	// 示例：保存到数据库
	// _, err := h.db.ExecContext(context.Background(), "INSERT INTO device_data ...", ...)
	// if err != nil {
	//     log.Printf("保存设备数据失败: %v", err)
	// }
}

func (h *KafkaMessageHandler) handleTransaction(msg *sarama.ConsumerMessage) {
	// 实际处理交易数据的逻辑
	log.Printf("处理交易数据: Topic=%s, Key=%s", msg.Topic, string(msg.Key))
}

// Stop 停止消息处理
func (h *KafkaMessageHandler) Stop() {
	h.cancel()
	log.Println("Kafka消息处理器已停止")
}
